package com.shujia.flink.window;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo1TimeWIndow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /*
         * java,1731052004000
         * java,1731052005000
         * java,1731052003000
         * java,1731052007000
         * java,1731052008000
         * java,1731052009000
         * java,1731052010000
         * java,1731052011000
         * java,1731052016000
         */
        DataStream<String> linesDS = env.socketTextStream("master", 8888);
        //解析数据
        DataStream<Tuple2<String, Long>> eventDS = linesDS
                .map(line -> {
                    String[] split = line.split(",");
                    String word = split[0];
                    //事件时间
                    long ts = Long.parseLong(split[1]);
                    return Tuple2.of(word, ts);
                }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        //指定时间字段和水位线
        DataStream<Tuple2<String, Long>> assDS = eventDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        //1、水位线等于最新一条数据的时间戳
                        //.<Tuple2<String, Long>>forMonotonousTimestamps()
                        //2、水位线减去5秒
                        .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        //指定时间字段
                        .withTimestampAssigner((kv, ts) -> kv.f1)
        );

        DataStream<Tuple2<String, Integer>> kvDS = assDS
                .map(kv -> Tuple2.of(kv.f0, 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT));

        /*
         * SlidingEventTimeWindows:滑动事件时间窗口
         * SlidingProcessingTimeWindows：滑动的处理时间窗口
         * TumblingEventTimeWindows滚动的事件时间窗口
         * TumblingProcessingTimeWindows：滚动的处理时间窗口
         */
        //统计最近5秒每个单词的数量
        DataStream<Tuple2<String, Integer>> countDS = kvDS.keyBy(kv -> kv.f0)
                //滚动的事件时间窗口
                .window(SlidingEventTimeWindows.of(Time.seconds(15), Time.seconds(5)))
                //统计单词的数量
                .reduce((kv1, kv2) -> Tuple2.of(kv1.f0, kv1.f1 + kv2.f1));

        countDS.print();

        env.execute();
    }
}
